package com.nami.gate.router.middleware.mq.producer;


import com.nami.gate.router.config.rocketmq.RocketMqConfig;
import com.nami.gate.router.middleware.mq.MqSender;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.UnsupportedEncodingException;
import java.util.List;

public class RocketMqSenderImpl implements MqSender {

    private static final Logger log = LoggerFactory.getLogger(RocketMqSenderImpl.class);

    private DefaultMQProducer producer;
    private RocketMqConfig config;

    public RocketMqSenderImpl(RocketMqConfig config) {
        this.config = config;
    }

    @Override
    public void start() throws Exception {
        producer = new DefaultMQProducer(config.getProducerGroup());
        producer.setNamesrvAddr(config.getNamesrvAddr());
        try {
            producer.start();
        } catch (Exception e) {
            e.printStackTrace();
            throw e;
        }
        log.info("[rocketmq] init success. namesrvAddr={}", config.getNamesrvAddr());
    }

    @Override
    public boolean send(String topic, String key, String msg) throws Exception {
        Message message = new Message();
        message.setTopic(topic);
        try {
            message.setBody(msg.getBytes("utf-8"));
            message.setKeys(key);
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }

        SendResult result = producer.send(message, queueSelector(topic, key));
        log.info("[mq-sender] topic:[{}] keys={} send msg success.", topic, key);
        return result.getSendStatus() == SendStatus.SEND_OK;
    }

    private MessageQueue queueSelector(String topic, String key) throws Exception {
        // 获取所有队列
        List<MessageQueue> messageQueues = producer.fetchPublishMessageQueues(topic);

        // 根据业务键计算哈希值，并选择队列
        int hashCode = key.hashCode();
        if (hashCode < 0) {
            hashCode = Math.abs(hashCode);
        }
        int index = hashCode % messageQueues.size();
        MessageQueue selectedQueue = messageQueues.get(index);
        log.info("[mq-sender] topic={}, key={} 当前队列数量：{}，queueId={}",
            topic, key, messageQueues.size(), selectedQueue.getQueueId());
        return selectedQueue;
    }

    @Override
    public boolean send(String topic, String msg) throws Exception {
       return this.send(topic, msg.getBytes("utf-8"));
    }

    @Override
    public boolean send(String topic, byte[] msg) throws Exception {
        Message message = new Message();
        message.setTopic(topic);
        message.setBody(msg);
        SendResult result = producer.send(message);
        return result.getSendStatus() == SendStatus.SEND_OK;
    }

    @Override
    public void shutdown() {
        if (producer != null) {
            producer.shutdown();
        }
    }


}
